package com.example.mq.mqserver;

import com.example.mq.common.Consumer;
import com.example.mq.common.MQException;
import com.example.mq.mqserver.core.*;
import com.example.mq.mqserver.datacenter.DiskDataCenter;
import com.example.mq.mqserver.datacenter.MemoryDataCenter;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

@Slf4j
@Getter
public class VirtualHost {
    // 虚拟主机名
    private final String virtualHostName;
    // 用于管理内存数据
    private final MemoryDataCenter memoryDataCenter = new MemoryDataCenter();
    // 用于管理硬盘数据
    private final DiskDataCenter diskDataCenter = new DiskDataCenter();
    // 用于 bandingKey 和 routingKey 的检验和匹配
    private final Router router = new Router();
    // 操作交换机的锁对象
    private final Object exchangeLocker = new Object();
    // 操作队列的锁对象
    private final Object queueLocker = new Object();
    // 管理消费者（消费者订阅队列，队列推送消息）
    private final ConsumerManager consumerManager = new ConsumerManager(this);

    public VirtualHost(String name) {
        this.virtualHostName = name;
        // 初始化：建库建表，初始化数据的插入
        diskDataCenter.init();
        try {
            // 恢复硬盘中的数据
            memoryDataCenter.recovery(diskDataCenter);
        } catch (IOException | MQException | ClassNotFoundException e) {
            log.info("恢复内存数据失败！");
            e.printStackTrace();
        }
    }

    /**
     * 1. 创建交换机
     * 如果交换机不存在，则创建，如果存在，就不创建
     * @param exchangeName  交换机名
     * @param exchangeType  交换机类型
     * @param durable       是否持久化
     * @param autoDelete    是否自动删除 (没人使用时)
     * @param arguments     参数设置
     * @return              创建成功/已经存在 返回 true，创建失败 返回 false
     */
    public boolean exchangeDeclare(String exchangeName, ExchangeType exchangeType,
                                   boolean durable, boolean autoDelete, Map<String, Object> arguments) {
        // 用来区分不同虚拟主机下的交换机
        exchangeName = virtualHostName + exchangeName;
        try {
            synchronized (exchangeLocker) {
                // 1. 判定该交换机是否已经存在，查内存中的就行
                Exchange existsExchange = memoryDataCenter.getExchange(exchangeName);
                if (existsExchange != null) {
                    log.info("交换机已经存在！exchangeName=" + exchangeName);
                    return true;
                }
                // 2. 创建交换机，构造 Exchange 对象
                Exchange exchange = new Exchange();
                exchange.setName(exchangeName);
                exchange.setType(exchangeType);
                exchange.setDurable(durable);
                exchange.setAutoDelete(autoDelete);
                exchange.setArguments(arguments);
                // 3. 交换机写入硬盘
                if (durable) {
                    diskDataCenter.insertExchange(exchange);
                }
                // 4. 写入内存
                memoryDataCenter.insertExchange(exchange);
                log.info("交换机创建完成！exchangeName=" + exchangeName);
            }
            return true;
        } catch (Exception e) {
            log.info("交换机创建失败！exchangeName=" + exchangeName);
            e.printStackTrace();
            return false;
        }
    }

    /**
     * 2. 删除交换机
     * @param exchangeName  交换机名
     * @return              删除成功返回 true，不存在/删除失败 返回 false
     */
    public boolean exchangeDelete(String exchangeName) {
        exchangeName = virtualHostName + exchangeName;
        try {
            synchronized (exchangeLocker) {
                // 1. 找到对应的交换机
                Exchange toDelete = memoryDataCenter.getExchange(exchangeName);
                if (toDelete == null) {
                    throw new MQException("[VirtualHost] 交换机不存在无法删除！");
                }
                // 2. 删除硬盘上的交换机
                if (toDelete.isDurable()) {
                    diskDataCenter.deleteExchange(exchangeName);
                }
                // 3. 删除内存中的交换机
                memoryDataCenter.deleteExchange(exchangeName);
                log.info("交换机删除成功！exchangeName=" + exchangeName);
            }
            return true;
        } catch (MQException e) {
            log.info("交换机删除失败！exchangeName=" + exchangeName);
            e.printStackTrace();
            return false;
        }
    }


    /**
     * 3. 创建队列
     * 队列不存在则创建
     * @param queueName     队列名
     * @param durable       是否持久化
     * @param exclusive     是否独占
     * @param autoDelete    自动删除
     * @param arguments     参数选项
     * @return              创建成功/已经存在返回 true，创建失败返回 false
     */
    public boolean queueDeclare(String queueName, boolean durable,
                                boolean exclusive, boolean autoDelete, Map<String, Object> arguments) {
        queueName = virtualHostName + queueName;
        try {
            synchronized (queueLocker) {
                // 1. 判定队列是否存在
                MQueue existsQueue = memoryDataCenter.getQueue(queueName);
                if (existsQueue != null) {
                    log.info("队列已经存在！queueName=" + queueName);
                    return true;
                }
                // 2. 构造队列
                MQueue queue = new MQueue();
                queue.setName(queueName);
                queue.setDurable(durable);
                queue.setExclusive(exclusive);
                queue.setAutoDelete(autoDelete);
                queue.setArguments(arguments);
                // 3. 写入硬盘
                if (durable) {
                    diskDataCenter.insertQueue(queue);
                }
                // 4. 写入硬盘
                memoryDataCenter.insertQueue(queue);
                log.info("队列创建成功！queueName=" + queueName);
            }
            return true;
        } catch (Exception e) {
            log.info("队列创建失败！queueName=" + queueName);
            e.printStackTrace();
            return false;
        }
    }

    //
    //

    /**
     * 4. 删除队列
     * @param queueName 队列名
     * @return          删除成功返回 true，创建失败/不存在返回 false
     */
    public boolean queueDelete(String queueName) {
        queueName = virtualHostName + queueName;
        try {
            synchronized (queueLocker) {
                // 1. 找到对应的队列
                MQueue toDelete = memoryDataCenter.getQueue(queueName);
                if (toDelete == null) {
                    throw new MQException("[VirtualHost] 队列不存在无法删除！");
                }
                // 2. 删除硬盘上的队列
                if (toDelete.isDurable()) {
                    diskDataCenter.deleteQueue(queueName);
                }
                // 3. 删除内存中的交换机
                memoryDataCenter.deleteQueue(queueName);
                log.info("队列删除成功！queueName=" + queueName);
            }
            return true;
        } catch (Exception e) {
            log.info("队列删除失败！queueName=" + queueName);
            e.printStackTrace();
            return false;
        }
    }


    /**
     * 5. 创建绑定
     * 绑定不存在,则创建
     * @param queueName     队列名
     * @param exchangeName  交换机名
     * @param bindingKey    bindingKey
     * @return              创建成功/已经存在 返回 true，创建失败 返回 false
     */
    public boolean queueBind(String queueName, String exchangeName, String bindingKey) {
        queueName = virtualHostName + queueName;
        exchangeName = virtualHostName + exchangeName;
        try {
            // 使用 exchangeLocker 和 queueLocker 加锁
            // 避免创建绑定时，对应的队列 或 交换机被删除
            synchronized (exchangeLocker) {
                synchronized (queueLocker) {
                    // 1. 判定当前的绑定是否已经存在
                    Binding existsBinding = memoryDataCenter.getBinding(exchangeName, queueName);
                    if (existsBinding != null) {
                        log.info("绑定已经存在！queueName=" + queueName + ", exchangeName=" + exchangeName);
                        return true;
                    }
                    // 2. 验证 bindingKey 是否合法
                    if (!router.checkBindingKey(bindingKey)) {
                        throw new MQException("[VirtualHost] bindingKey 非法！bindingKey=" + bindingKey);
                    }
                    // 3. 构造绑定对象
                    Binding binding = new Binding();
                    binding.setExchangeName(exchangeName);
                    binding.setQueueName(queueName);
                    binding.setBindingKey(bindingKey);
                    // 4. 获取对应的交换机和队列，如果交换机或队列不存在，那么绑定也不能创建
                    MQueue queue = memoryDataCenter.getQueue(queueName);
                    if (queue == null) {
                        throw new MQException("[VirtualHost] 队列不存在！queueName=" + queueName);
                    }
                    Exchange exchange = memoryDataCenter.getExchange(exchangeName);
                    if (exchange == null) {
                        throw new MQException("[VirtualHost] 交换机不存在！exchangeName=" + exchangeName);
                    }
                    // 5. 将绑定写入硬盘
                    if (queue.isDurable() && exchange.isDurable()) {
                        diskDataCenter.insertBinding(binding);
                    }
                    // 6. 将绑定写入内存
                    memoryDataCenter.insertBinding(binding);
                    log.info("创建绑定成功！queueName=" + queueName + ", exchangeName=" + exchangeName);
                }
            }
            return true;
        } catch(MQException e) {
            log.info("创建绑定失败！queueName=" + queueName + ", exchangeName=" + exchangeName);
            e.printStackTrace();
            return false;
        }
    }

    /**
     * 6. 删除绑定
     * @param queueName     队列名
     * @param exchangeName  交换机名
     * @return              删除成功返回 true，删除失败/不存在 返回 false
     */
    public boolean queueUnbind(String queueName, String exchangeName) {
        queueName = virtualHostName + queueName;
        exchangeName = virtualHostName + exchangeName;
        try {
            // 两把锁的加锁顺序要统一，否则可能出现死锁
            synchronized (exchangeLocker) {
                synchronized (queueLocker) {
                    // 1. 获取 binding 看是否存在
                    Binding binding = memoryDataCenter.getBinding(exchangeName, queueName);
                    if (binding == null) {
                        throw new MQException("[VirtualHost] 绑定不存在！queueName=" + queueName + ", exchangeName=" + exchangeName);
                    }
                    // 不判断交换机或队列是否存在，都进行删除
                    // 数据库中删除没有副作用，内存中最多抛出一个MqException
                    // 2. 不管交换机和队列有没有持久化。都尝试从数据库中删除
                    diskDataCenter.deleteBinding(binding);
                    // 3. 删除内存的数据
                    memoryDataCenter.deleteBinding(binding);
                    log.info("删除绑定成功！queueName=" + queueName + ", exchangeName=" + exchangeName);
                }
            }
            return true;
        } catch (Exception e) {
            log.info("删除绑定失败！queueName=" + queueName + ", exchangeName=" + exchangeName);
            e.printStackTrace();
            return false;
        }
    }


    /**
     * 7. 发送消息
     * @param exchangeName      交换机名
     * @param routingKey        routingKey
     * @param basicProperties   消息的基本属性
     * @param body              消息正文
     * @return                  发送成功返回 true，发送失败返回 false
     */
    public boolean basicPublish(String exchangeName, String routingKey, BasicProperties basicProperties, byte[] body) {
        try {
            // 1. 转换交换机的名字
            exchangeName = virtualHostName + exchangeName;
            // 2. 检查 routingKey 是否合法
            if (!router.checkRoutingKey(routingKey)) {
                throw new MQException("[VirtualHost] routingKey 非法！routingKey=" + routingKey);
            }
            // 3. 查找交换机对象
            Exchange exchange = memoryDataCenter.getExchange(exchangeName);
            if (exchange == null) {
                throw new MQException("[VirtualHost] 交换机不存在！exchangeName=" + exchangeName);
            }
            // 4. 判定交换机的类型
            if (exchange.getType() == ExchangeType.DIRECT) {
                // 按照 直接交换机 的方式转发消息
                // 此时 routingKey 就是指定的队列，消息转发给 queueName 为 routingKey 的队列
                // 此时，可以无视绑定关系（不管交换机 和 队列有没有关系）
                String queueName = virtualHostName + routingKey;
                // 5. 构造一个消息对象
                Message message = Message.createMessageWithId(routingKey, basicProperties, body);
                // 6. 查找指定的队列对象
                MQueue queue = memoryDataCenter.getQueue(queueName);
                if (queue == null) {
                    throw new MQException("[VirtualHost] 队列不存在！queueName=" + queueName);
                }
                // 7. 消息发送给队列
                sendMessage(queue, message);
            } else {
                // 按照 扇出交换机 和 主题交换机 的方式转发消息
                // 5. 找到该交换机关联的所有绑定，并且遍历所有的绑定对象
                ConcurrentHashMap<String, Binding> bindingsMap = memoryDataCenter.getBindings(exchangeName);
                for (Map.Entry<String, Binding> entry : bindingsMap.entrySet()) {
                    // 1) 获取到绑定对象，判定对应的队列是否存在
                    Binding binding = entry.getValue();
                    MQueue queue = memoryDataCenter.getQueue(binding.getQueueName());
                    if (queue == null) {
                        log.info("basicPublish 发送消息时，发现队列不存在！queueName=" + binding.getQueueName());
                        continue;
                    }
                    // 2) 构造消息对象
                    Message message = Message.createMessageWithId(routingKey, basicProperties, body);
                    // 3) 判定该消息能否转发给该队列
                    //	  如果是 FANOUT，所有绑定的队列都要转发
                    //	  如果是 TOPIC，需要判定消息的 routingKey 和绑定的 bindingKey 是否一致
                    if (!router.route(exchange.getType(), binding, message)) {
                        continue;
                    }
                    // 4) 消息发送给队列
                    sendMessage(queue, message);
                }
            }
            return true;
        } catch (Exception e) {
            log.info("消息发送失败！");
            e.printStackTrace();
            return false;
        }
    }

    // 将消息发送给队列（消息写入硬盘和内存里）
    private void sendMessage(MQueue queue, Message message) throws IOException, MQException, InterruptedException {
        int deliverMode = message.getDeliverMode();
        // deliverMode 为 1 表示 不持久化，为 2 表示 要持久化
        // 1. 写入硬盘
        if (deliverMode == 2) {
            // 如果该队列不持久化，那么队列对应的目录和文件就不存在，消息持久化时就抛出异常
            diskDataCenter.sendMessage(queue, message);
        }
        // 2. 写入内存
        memoryDataCenter.sendMessage(queue, message);

        // 将队列名放入阻塞队列中，扫描线程获取阻塞队列中的队列名，然后找到指定队列，获取信息交给线程池，消费消息
        consumerManager.notifyConsume(queue.getName());
    }


    /**
     * 8. 订阅消息
     * 消费者订阅队列，队列收到消息后，就会将消息推送给消费者
     *
     * @param consumerTag   消费者身份标识
     * @param queueName     队列名
     * @param autoAck       消费者消费完成后，应答的方式，为 true 表示自动应答，false 表示手动应答
     * @param consumer      回调函数，此处类型设定为函数式接口，后续调用 basicConsume 传递参数时，可以写成 lambda 表达式
     * @return              订阅成功：true，订阅失败：false
     */
    public boolean basicConsume (String consumerTag, String queueName, boolean autoAck, Consumer consumer) {
        queueName = virtualHostName + queueName;
        try {
            // 构造一个 ConsumerEnv 对象，把对应的队列找到，把这个 ConsumerEnv 对象加入队列的订阅表中。
            consumerManager.addConsumer(consumerTag, queueName, autoAck, consumer);
            log.info("basicConsume 成功！queueName=" + queueName);
            return true;
       } catch (MQException e) {
            log.info("basicConsume 失败！queueName=" + queueName);
            e.printStackTrace();
            return false;
        }
    }

    /**
     * 9. 确认消息
     * 消费者消费完成之后，手动确认消息
     * 消息确认后，就可以删除了
     * @param queueName 队列名
     * @param messageId 消息Id
     * @return  确认成功：true，确认失败：false
     */
    public boolean basicAck(String queueName, String messageId) {
        queueName = virtualHostName + queueName;
        try {
            // 1. 获取消息和队列
            Message message = memoryDataCenter.getMessage(messageId);
            if (message == null) {
                throw new MQException("[VirtualHost] 要确认的消息不存在！messageId=" + messageId);
            }
            MQueue queue = memoryDataCenter.getQueue(queueName);
            if (queue == null) {
                throw new MQException("[VirtualHost] 要确认的队列不存在！queueName=" + queueName);
            }
            // 2. 删除硬盘上的消息
            if (message.getDeliverMode() == 2) {
                diskDataCenter.deleteMessage(queue, message.getMessageId(), memoryDataCenter);
            }
            // 3. 删除内存的消息中心上的消息
            memoryDataCenter.removeMessage(message.getMessageId());
            // 4. 删除待确认消息表的消息
            memoryDataCenter.removeMessageWaitAck(queue.getName(), message.getMessageId());
            log.info("basicAck 成功！消息被成功确认！queueName=" + queueName + ", messageId=" + messageId);
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            log.info("basicAck 失败！消息确认失败！queueName=" + queueName + ", messageId=" + messageId);
            return false;
        }
    }
}
